package cn.doitedu.flink.demos.drools;

import cn.doitedu.flink.demos.SourcesTest;
import org.apache.commons.io.FileUtils;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.internal.InternalValueState;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.kie.api.io.ResourceType;
import org.kie.api.runtime.KieSession;
import org.kie.internal.utils.KieHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.util.*;

/**
 * 利用drools来实现动态规则处理
 *
 */
public class DynamicDroolsRules {
    public static void main(String[] args) throws Exception {

        Configuration config = new Configuration();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        /**
         * 读取规则流
         */
        String ruleStr = FileUtils.readFileToString(new File("D:\\IdeaProjects\\hunter_teach\\flink_tech\\src\\main\\resources\\count_limit.drl"), "utf-8");
        DataStreamSource<String> rule = env.fromCollection(Arrays.asList(ruleStr));

        /**
         * 读取事件流，并按key分组
         */
        KeyedStream<DynamicDroolsEvent, String> source = env.addSource(SourcesTest.createSocketSource("localhost", 9999))
                .map(s -> {
                    String[] arr = s.split(",");
                    return new DynamicDroolsEvent(arr[0], arr[1], Long.parseLong(arr[2]), null);
                }).returns(DynamicDroolsEvent.class)
                .keyBy(new KeySelector<DynamicDroolsEvent, String>() {

                    @Override
                    public String getKey(DynamicDroolsEvent value) throws Exception {
                        return value.getUid();
                    }
                });

        MapStateDescriptor<String, String> ruleDesc = new MapStateDescriptor<>("rule_desc", String.class, String.class);
        BroadcastStream<String> broadcast = rule.broadcast(ruleDesc);

        ValueStateDescriptor vStateDesc = new ValueStateDescriptor<>("events", TypeInformation.of(new TypeHint<List<DynamicDroolsEvent>>() {}));

        BroadcastConnectedStream<DynamicDroolsEvent, String> connect = source.connect(broadcast);
        connect.process(new KeyedBroadcastProcessFunction<String, DynamicDroolsEvent, String, String>() {

            ValueState<List<DynamicDroolsEvent>> vstate;
            @Override
            public void open(Configuration parameters) throws Exception {

                vstate = getRuntimeContext().getState(vStateDesc);
            }

            @Override
            public void processElement(DynamicDroolsEvent value, ReadOnlyContext ctx, Collector<String> out) throws Exception {
                String testRule = ctx.getBroadcastState(ruleDesc).get("test");

                List<DynamicDroolsEvent> elist = vstate.value();
                // 如果获取的状态为null，则初始化一个list
                if(elist == null) elist = new ArrayList<DynamicDroolsEvent>();
                elist.add(value);

                System.out.println("当前event list 的size为： " + elist.size());
                Iterator<DynamicDroolsEvent> iterator = elist.iterator();
                while(iterator.hasNext()){
                    DynamicDroolsEvent e = iterator.next();
                    if(value.getTimeStamp() - e.getTimeStamp() > 4000) {
                        iterator.remove();
                        System.out.println("移除过期event");
                    }
                }
                System.out.println("当前event list的size为： " + elist.size());

                // 更新list到状态中
                vstate.update(elist);


                /**
                 *  进行规则判断处理
                 */
                KieHelper helper = new KieHelper();
                helper.addContent(testRule, ResourceType.DRL);
                KieSession kSession = helper.build().newKieSession();

                kSession.insert(elist);
                int count = kSession.fireAllRules();
                System.out.println("命中的规则条数为： " + count);

                kSession.dispose();
                kSession.destroy();
            }

            @Override
            public void processBroadcastElement(String value, Context ctx, Collector<String> out) throws Exception {
                ctx.getBroadcastState(ruleDesc).put("test",value);
                System.out.println("放入一条规则");
            }
        }).print();



        env.execute();

    }
}
